﻿using System;
using System.Collections.Generic;
using System.Collections.ObjectModel;
using System.Linq;
using System.Text;
using System.IO;
using System.Threading;
using System.Diagnostics.CodeAnalysis;
using System.Diagnostics;
using System.Xml.Linq;
using ThreeByte.ActiveDeck.Publish.Model;

namespace ThreeByte.ActiveDeck.Publish
{
    [SuppressMessage("Microsoft.Design", "CA1001:TypesThatOwnDisposableFieldsShouldBeDisposable",
        Justification = "Timer and channelsLock do not cause a memory leak because ActivePublisher is a singleton in an app domain.")]
    public partial class ActivePublisher : IPublisher
    {

        int maxMessageId = 0;
        int currentNumberMessages = 0;
        TimeSpan ttl;
        int maxNumberMessages;
        object syncRoot = new object();
        Dictionary<string, PublishedChannel> Channels = new Dictionary<string, PublishedChannel>();
        ReaderWriterLockSlim channelsLock = new ReaderWriterLockSlim();
        Timer timer;

        public static readonly ActivePublisher Instance = new ActivePublisher();

        //Make this a real singleton
        private ActivePublisher()
        {
            //TODO: Make these configurable
            this.ttl = TimeSpan.FromSeconds(300);
            this.maxNumberMessages = 1024 * 1024;
            this.timer = new Timer(new TimerCallback(this.PruneStaleChannels), null, (int)this.ttl.TotalMilliseconds, Timeout.Infinite);
        }

        public IAsyncResult BeginPublishMessage(string channel, XElement body, AsyncCallback callback, object state)
        {
            if (this.currentNumberMessages > this.maxNumberMessages)
            {
                throw new InvalidOperationException("Pub/sub backend cannot accept new messages bacause the memory store quota has been reached.");
            }
            return new PublishMessageAsyncResult(this, channel, body, callback, state);
        }

        public int EndPublishMessage(IAsyncResult result)
        {
            return PublishMessageAsyncResult.End(result);
        }

        public IAsyncResult BeginSubscribe(IDictionary<string, int> subscriptions, TimeSpan timeout, AsyncCallback callback, object state)
        {
            return new SubscriptionAsyncResult(this, subscriptions, timeout, callback, state);
        }

        public IEnumerable<ActiveMessage> EndSubscribe(IAsyncResult result)
        {
            return SubscriptionAsyncResult.End(result);
        }

#if CONSISTENCY
        void AssertDatabaseConsistency()
        {
            this.channelsLock.EnterReadLock();
            try
            {
                int bookmarks = 0;
                int messages = 0;
                foreach (KeyValuePair<int, PublishedChannel> entry in this.channels)
                {
                    entry.Value.Lock.EnterReadLock();
                    try
                    {
                        bookmarks += entry.Value.Bookmarks.Count;
                        messages += entry.Value.Messages.Count;
                        int last = entry.Value.Messages.Count > 0 ? entry.Value.Messages.Last().ActiveMessage.MessageId : -1;
                        if (entry.Value.Bookmarks.Any(b => b.From <= last))
                        {
                            Console.WriteLine("Consistency check failed: channel={0}, bookmarks={1}, messages={2}",
                                entry.Key, entry.Value.Bookmarks.Count, entry.Value.Messages.Count);
                        }
                    }
                    finally
                    {
                        entry.Value.Lock.ExitReadLock();
                    }
                }
                Console.WriteLine("channels = {0}, bookmarks = {1}, messages = {2}, memory = {3}",
                    this.channels.Count, bookmarks, messages, this.currentSize);
            }
            finally
            {
                this.channelsLock.ExitReadLock();
            }
        }
#endif


        public IAsyncResult BeginGetChannels(string prefix, AsyncCallback callback, object state) {

            Dictionary<string, int> channelsToReturn = new Dictionary<string, int>();

            this.channelsLock.EnterReadLock();
            try {
                foreach(KeyValuePair<string, PublishedChannel> entry in this.Channels) {
                    if(entry.Key.StartsWith(prefix)) {
                        int messageId = 0;
                        entry.Value.Lock.EnterReadLock();

                        try {
                            PublishedMessage m = entry.Value.Messages.OrderByDescending(p => p.ActiveMessage.MessageId).FirstOrDefault();
                            if(m != null) {
                                messageId = m.ActiveMessage.MessageId;
                            }
                        } finally {
                            entry.Value.Lock.ExitReadLock();
                        }

                        channelsToReturn[entry.Key] = messageId;
                    }
                }
            } finally {
                this.channelsLock.ExitReadLock();
            }

            return new CompletedAsyncResult<Dictionary<string, int>>(channelsToReturn, callback, state);
        }

        public Dictionary<string, int> EndGetChannels(IAsyncResult result) {
            CompletedAsyncResult<Dictionary<string, int>> channelResult = (CompletedAsyncResult<Dictionary<string, int>>)result;
            return channelResult.Result;            
        }


        int PruneExpiredMessages(PublishedChannel publishedChannel, DateTime now)
        {
            int totalSizeReduction = 0;
            while (publishedChannel.Messages.Count > 0 && publishedChannel.Messages[0].Expire < now)
            {
                totalSizeReduction++;
                publishedChannel.Messages.RemoveAt(0);
            }
            return totalSizeReduction;
        }

        void PruneStaleChannels(object state)
        {
            DateTime now = DateTime.Now;
            List<string> removeCandidates = new List<string>();
            this.channelsLock.EnterUpgradeableReadLock();
            try
            {
                foreach (KeyValuePair<string, PublishedChannel> entry in this.Channels)
                {
                    bool pruneExpiredMessages = false;
                    entry.Value.Lock.EnterReadLock();
                    try
                    {
                        if (this.IsChannelStale(entry.Value, now))
                        {
                            removeCandidates.Add(entry.Key);
                        }
                        else
                        {
                            pruneExpiredMessages = entry.Value.Messages.Count > 0 && entry.Value.Messages[0].Expire < now;
                        }
                    }
                    finally
                    {
                        entry.Value.Lock.ExitReadLock();
                    }
                    if (pruneExpiredMessages)
                    {
                        int totalSizeReduction;
                        entry.Value.Lock.EnterWriteLock();
                        try
                        {
                            totalSizeReduction = this.PruneExpiredMessages(entry.Value, now);
                        }
                        finally
                        {
                            entry.Value.Lock.ExitWriteLock();
                        }
                        if (totalSizeReduction > 0)
                        {
                            lock (this.syncRoot)
                            {
                                this.currentNumberMessages -= totalSizeReduction;
                            }
                        }
                    }
                }
                if (removeCandidates.Count > 0)
                {
                    this.channelsLock.EnterWriteLock();
                    try
                    {
                        foreach (string channel in removeCandidates)
                        {
                            PublishedChannel publishedChannel = this.Channels[channel];
                            if (this.IsChannelStale(publishedChannel, now))
                            {
                                int totalSizeReduction = publishedChannel.Messages.Count;
                                lock (this.syncRoot)
                                {
                                    this.currentNumberMessages -= totalSizeReduction;
                                }
                                publishedChannel.Lock.Dispose();
                                this.Channels.Remove(channel);
                            }
                        }
                    }
                    finally
                    {
                        this.channelsLock.ExitWriteLock();
                    }
                }
            }
            finally
            {
                this.channelsLock.ExitUpgradeableReadLock();
            }
            this.timer.Change((int)this.ttl.TotalMilliseconds, Timeout.Infinite);
#if CONSISTENCY
            this.AssertDatabaseConsistency();
#endif
        }

        bool IsChannelStale(PublishedChannel publishedChannel, DateTime now)
        {
            if (publishedChannel.Bookmarks.Count > 0)
            {
                return false;
            }
            if (publishedChannel.Messages.Count > 0 && publishedChannel.Messages[publishedChannel.Messages.Count - 1].Expire > now)
            {
                return false;
            }
            return true;
        }

        class PublishedChannel
        {
            public ReaderWriterLockSlim Lock { get; private set; }
            public Collection<PublishedMessage> Messages { get; private set; }
            public Collection<Bookmark> Bookmarks { get; private set; }

            public PublishedChannel()
            {
                this.Lock = new ReaderWriterLockSlim();
                this.Messages = new Collection<PublishedMessage>();
                this.Bookmarks = new Collection<Bookmark>();
            }
        }

        class PublishedMessage
        {
            public DateTime Expire { get; set; }
            public ActiveMessage ActiveMessage { get; set; }
        }

        class Bookmark
        {
            public int From { get; set; }
            public SubscriptionAsyncResult Continuation { get; set; }
            public PublishedChannel PublishedChannel { get; set; }
        }

    }
}
